use std::collections::HashMap;
use std::ffi::{OsStr, OsString};
use std::io::SeekFrom;
use std::net::SocketAddr;
use std::os::unix::fs::MetadataExt;
use std::path::{Path, PathBuf};
use std::sync::Arc;
use std::time::SystemTime;

use tokio::sync::{Mutex, RwLock};
use tokio::fs::File;
use crc::{Crc, CRC_16_IBM_SDLC};
use fuse3::raw::prelude::ReplyStatFs;
use tokio::io::{AsyncReadExt, AsyncSeekExt, AsyncWriteExt};
use xor_name::XorName;

use tracing::{debug, error, info, trace, warn};
use crate::Attribute;

use crate::error::*;
use crate::remote::NodeRemote;
use crate::metadata::{NTree, NodeAttr, Location};
use crate::network::Event;
use crate::storage::LocalStorage;

pub const X25: Crc<u16> = Crc::<u16>::new(&CRC_16_IBM_SDLC);


pub enum OpenLocation {
    Local((u64, RwLock<File>)),
    Remote((u64, XorName)),
}

pub struct Scheduler {
    generator: Arc<Mutex<snowflake::SnowflakeIdGenerator>>,
    metadata: Arc<RwLock<NTree>>,
    remote: Arc<RwLock<HashMap<XorName, NodeRemote>>>,
    opened: Arc<RwLock<HashMap<u64, OpenLocation>>>,
    local: Arc<LocalStorage>,
}

impl Scheduler {
    pub async fn setup(root: Vec<PathBuf>, local: SocketAddr) -> Result<Self, FileSystemError> {
        let metadata = NTree::load(root.clone()).await?;
        let ip = local.ip().to_string();
        let machine = X25.checksum(&ip.as_bytes());
        let node = local.port() as i32;
        let snow = snowflake::SnowflakeIdGenerator::new(machine as i32, node);
        let local = LocalStorage::new(root);
        let scheduler = Scheduler {
            generator: Arc::new(Mutex::new(snow)),
            metadata: Arc::new(RwLock::new(metadata)),
            remote: Arc::new(RwLock::new(HashMap::new())),
            opened: Arc::new(RwLock::new(HashMap::new())),
            local: Arc::new(local),
        };
        Ok(scheduler)
    }

    pub async fn store(&self, peer: XorName, data: Vec<(PathBuf, Attribute)>) {
        let exists = self.remote.read().await.contains_key(&peer);
        if exists {
            for ( path, attribute) in data {
                let attr = NodeAttr::File(Location::Remote(peer), attribute);
                match self.metadata.write().await.store(&path, attr).await {
                    Some(id) => trace!("store remote, id {}, path {}", id, path.display()),
                    None => {
                        warn!("store remote, node exists {}, {}", peer, path.display());
                    },
                }
            }
        } else {
            warn!("remote not found {}", peer)
        }
    }

    pub async fn store_one(&self, peer: XorName, data: (PathBuf, Attribute)) -> Result<u64, FileSystemError> {
        let exists = self.remote.read().await.contains_key(&peer);
        if exists {
            let attr = NodeAttr::File(Location::Remote(peer), data.1);
            if let Some(id) = self.metadata.write().await.store(&data.0, attr).await {
                Ok(id)
            } else {
                Err(FileSystemError::TreeNodeExistsError(data.0))
            }
        } else {
            Err(FileSystemError::RemoteNotFoundError(peer))
        }
    }

    pub fn generator(&self) -> Arc<Mutex<snowflake::SnowflakeIdGenerator>> {
        self.generator.clone()
    }

    pub async fn add_remote(&self, peer: XorName, remote: NodeRemote) {
        let array = self.remote.read().await.iter().map(|(_k, v)| v.address()).collect::<Vec<SocketAddr>>();
        remote.contacts(array).await;
        let array = self.metadata.read().await.to_array().await;
        remote.metadata(array).await;
        self.remote.write().await.insert(peer, remote);
    }

    pub async fn have_remote(&self, peer: &XorName) -> bool {
        match self.remote.read().await.get(peer) {
            None => false,
            Some(node) => node.running()
        }
    }

    pub async fn send_remote(&self, peer: XorName, event: Event) -> Result<(), FileSystemError> {
        trace!("send remote {}, event {:?}", peer, event);
        match self.remote.read().await.get(&peer) {
            None => Err(FileSystemError::RemoteNotFoundError(peer)),
            Some(remote) => remote.send(event).await
        }
    }

    pub async fn broadcast(&self, event: Event) -> Result<(), FileSystemError> {
        trace!("broadcast remote {:?}", &event);
        for (_peer, remote) in self.remote.read().await.iter() {
            if let Err(e) = remote.send(event.clone()).await {
                error!("broadcast remote error: {:?}. {}", event, e);
            }
        }
        Ok(())
    }

    pub async fn lookup(&self, parent: u64, name: &OsStr) -> Option<(u64, NodeAttr)> {
        if let Some(node) = self.metadata.read().await.lookup(parent, name).await {
            let id = node.read().await.id();
            let attr = node.read().await.node_attr();
            Some((id, attr))
        } else {
            None
        }
    }

    pub async fn locate(&self, id: u64) -> Option<NodeAttr> {
        if let Some(node) = self.metadata.read().await.locate(id).await {
            Some(node.read().await.node_attr())
        } else {
            None
        }
    }

    pub async fn mkdir(&self, parent: u64, name: &OsStr) -> Result<u64, FileSystemError> {
        let locate = self.metadata.read().await.locate(parent).await;
        if let Some(node) = locate {
            let path = node.read().await.path().join(name);
            let _ = self.local.mkdir(&path).await?;
            if let Some(node) = self.metadata.write().await.store(&path, NodeAttr::Directory).await {
                info!("mkdir local {}", path.display());
                Ok(node)
            } else {
                Err(FileSystemError::TreeNodeExistsError(path))
            }
        } else {
            Err(FileSystemError::NodeNotFoundError(parent))
        }
    }

    pub async fn create(&self, parent: u64, name: &OsStr) -> Result<(u64, u64, NodeAttr), FileSystemError> {
        let locate = self.metadata.read().await.locate(parent).await;
        if let Some(node) = locate {
            let path = node.read().await.path().join(name);
            // create file
            let (root, file) = self.local.create(&path).await?;
            let metadata = file.metadata().await?;
            let attribute = Attribute::from(metadata);
            let attr = NodeAttr::File(Location::Local(root), attribute.clone());
            // add metadata
            let created = self.metadata.write().await.store(&path, attr.clone()).await;
            if let Some(node) = created {
                let key = self.generator.lock().await.generate() as u64;
                // add opened
                self.opened.write().await.insert(key, OpenLocation::Local((node, RwLock::new(file))));
                let event = Event::Create((path.clone(), attribute.clone()));
                // broadcast event
                self.broadcast(event).await?;
                info!("create node local {}", path.display());
                Ok((node, key, attr))
            } else {
                Err(FileSystemError::TreeNodeExistsError(path))
            }
        } else {
            Err(FileSystemError::NodeNotFoundError(parent))
        }
    }

    pub async fn remove_local(&self, parent: u64, name: &OsStr) -> Result<u64, FileSystemError> {
        let (parent, removed) = self.metadata.write().await.remove_node(parent, name).await?;
        let attr = removed.read().await.node_attr();
        let path = removed.read().await.path();
        let id = removed.read().await.id();
        match attr {
            NodeAttr::Directory => {
                warn!("remove local from metadata, {} is not a file.", path.display());
                parent.write().await.add_child(removed);
                return Err(FileSystemError::NodeNotFileError(path))
            }
            NodeAttr::File(location, ..) => {
                info!("remove local from metadata: {}, {:?}", path.display(), location);
                if let Location::Local(root) = location {
                    if let Err(e) = self.local.remove(root, &path).await {
                        warn!("remove local from local file error {}", e);
                        parent.write().await.add_child(removed);
                        return Err(e)
                    }
                }
                let event = Event::Remove(path);
                self.broadcast(event).await?;
            }
        }
        Ok(id)
    }

    pub async fn remove_remote(&self, _peer: XorName, path: &Path) -> Result<u64, FileSystemError> {
        let (parent, removed) = self.metadata.write().await.remove_path(path).await?;
        let attr = removed.read().await.node_attr();
        let path = removed.read().await.path();
        let id = removed.read().await.id();
        match attr {
            NodeAttr::Directory => {
                warn!("remove remote from metadata, {} is not a file.", path.display());
                parent.write().await.add_child(removed);
                return Err(FileSystemError::NodeNotFileError(path))
            }
            NodeAttr::File(location, ..) => {
                info!("remove remote from metadata: {}, {:?}", path.display(), location);
                if let Location::Local(root) = location {
                    if let Err(e) = self.local.remove(root, &path).await {
                        warn!("remove remote from local file error {}", e);
                        parent.write().await.add_child(removed);
                        return Err(e)
                    }
                }
            }
        }
        Ok(id)
    }

    pub async fn rmdir_local(&self, parent: u64, name: &OsStr) -> Result<u64, FileSystemError> {
        let (parent, removed) = self.metadata.write().await.remove_node(parent, name).await?;
        let attr = removed.read().await.node_attr();
        let path = removed.read().await.path();
        let id = removed.read().await.id();
        match attr {
            NodeAttr::Directory => {
                info!("rmdir local from metadata: {}", path.display());
                if let Err(e) = self.local.rmdir(&path).await {
                    warn!("rmdir local from local error {}", e);
                }
                let event = Event::Rmdir(path);
                self.broadcast(event).await?;
            }
            NodeAttr::File(..) => {
                warn!("rmdir local from metadata, {} is not a directory.", path.display());
                parent.write().await.add_child(removed);
                return Err(FileSystemError::NodeNotFileError(path))
            }
        }
        Ok(id)
    }

    pub async fn rmdir_remote(&self, _peer: XorName, path: &Path) -> Result<u64, FileSystemError> {
        let (parent, removed) = self.metadata.write().await.remove_path(path).await?;
        let attr = removed.read().await.node_attr();
        let path = removed.read().await.path();
        let id = removed.read().await.id();
        match attr {
            NodeAttr::Directory => {
                info!("rmdir remote from metadata: {}", path.display());
                if let Err(e) = self.local.rmdir(&path).await {
                    warn!("rmdir local from local error {}", e);
                }
            }
            NodeAttr::File(..) => {
                warn!("rmdir remote from metadata, {} is not a directory.", path.display());
                parent.write().await.add_child(removed);
                return Err(FileSystemError::NodeNotFileError(path))
            }
        }
        Ok(id)
    }

    pub async fn rename_local(&self, parent: u64, name: &OsStr, new_parent: u64, new_name: &OsStr) -> Result<(), FileSystemError> {
        let (source, target, node) = self.metadata.write().await.rename_node(parent, name, new_parent, new_name).await?;
        let attr = node.read().await.node_attr();
        match attr {
            NodeAttr::Directory => {
                let _ = self.local.mvdir(&source, &target).await?;
                info!("rename dir local, source: {}, target: {}", source.display(), target.display());
            }
            NodeAttr::File(location, ..) => {
                if let Location::Local(root) = location {
                    let _ = self.local.rename(root, &source, &target).await?;
                }
                info!("rename file local, source: {}, target: {}", source.display(), target.display());
            }
        }
        let event = Event::Rename {source, target};
        self.broadcast(event).await?;
        Ok(())
    }

    pub async fn move_remote(&self, _peer: XorName, source: &Path, target: &Path) -> Result<(), FileSystemError> {
        info!("rename node remote, source: {}, target: {}", source.display(), target.display());
        let node = self.metadata.write().await.rename_path(source, target).await?;
        let attr = node.read().await.node_attr();
        match attr {
            NodeAttr::Directory => {
                self.local.mvdir(source, target).await
            }
            NodeAttr::File(location, ..) => {
                if let Location::Local(root) = location {
                    self.local.rename(root, source, target).await
                } else {
                    Ok(())
                }
            }
        }
    }

    pub async fn open_path(&self, path: &Path) -> Result<u64, FileSystemError> {
        let locate = self.metadata.read().await.r#match(path).await;
        if let Some(node) = locate {
            let id = node.read().await.id();
            let attr = node.read().await.node_attr();
            self.open(id, &path, attr).await
        } else {
            Err(FileSystemError::PathNotFoundError(path.to_path_buf()))
        }
    }

    pub async fn open_node(&self, id: u64) -> Result<u64, FileSystemError> {
        let locate = self.metadata.read().await.locate(id).await;
        if let Some(node) = locate {
            let lock = node.read().await;
            let id = lock.id();
            let path = lock.path();
            let attr = lock.node_attr();
            debug!("open node {}", path.display());
            self.open(id, &path, attr).await
        } else {
            Err(FileSystemError::NodeNotFoundError(id))
        }
    }

    async fn open(&self, id: u64, path: &Path, attr: NodeAttr) -> Result<u64, FileSystemError> {
        match attr {
            NodeAttr::File(location, _attr) => {
                match location {
                    Location::Local(root) => {
                        trace!("open from local {}", id);
                        let key = self.generator.lock().await.generate() as u64;
                        let file = self.local.open(root, path).await?;
                        self.opened.write().await.insert(key, OpenLocation::Local((id, RwLock::new(file))));
                        Ok(key)
                    }
                    Location::Remote(peer) => {
                        trace!("open from remote {}, {}", id, peer);
                        match self.remote.read().await.get(&peer) {
                            None => Err(FileSystemError::RemoteNotFoundError(peer)),
                            Some(v) => {
                                let fh = v.open(path).await?;
                                self.opened.write().await.insert(fh, OpenLocation::Remote((id, peer)));
                                Ok(fh)
                            }
                        }
                    }
                }
            },
            NodeAttr::Directory => Err(FileSystemError::NodeNotFileError(path.to_path_buf()))
        }
    }

    pub async fn read(&self, _id: u64, fh: u64, offset: u64, size: u32) -> Result<Vec<u8>, FileSystemError> {
        match self.opened.read().await.get(&fh) {
            None => Err(FileSystemError::UnknownFileError(fh)),
            Some(d) => {
                match d {
                    OpenLocation::Local((node, file)) => {
                        let mut f = file.write().await;
                        trace!("read from local, node {}, fh {}", node, fh);
                        let length = f.metadata().await.unwrap().size();
                        let length = if (offset + size as u64) > length {
                            length - offset
                        } else {
                            size as u64
                        };
                        if length > 0 {
                            let mut buf = vec![0u8; length as usize];
                            f.seek(SeekFrom::Start(offset)).await?;
                            f.read_exact(&mut buf).await?;
                            Ok(buf)
                        } else {
                            Ok(vec![])
                        }
                    }
                    OpenLocation::Remote((node, peer)) => {
                        debug!("read from remote, node {}, fh {}, peer {}", node, fh, peer);
                        let now = SystemTime::now();
                        match self.remote.read().await.get(&peer) {
                            None => Err(FileSystemError::RemoteNotFoundError(peer.clone())),
                            Some(v) => {
                                debug!("read from remote, time use {}", now.elapsed().unwrap().as_nanos());
                                v.read(fh, offset, size).await
                            }
                        }
                    }
                }
            }
        }
    }

    pub async fn write(&self, _id: u64, fh: u64, offset: u64, data: &[u8]) -> Result<u32, FileSystemError> {
        if let Some(d) = self.opened.read().await.get(&fh) {
            match d {
                OpenLocation::Local((node, file)) => {
                    let mut f = file.write().await;
                    let size = data.len();
                    trace!("write from local, node {}, fh {}, length {}", node, fh, size);
                    f.seek(SeekFrom::Start(offset)).await?;
                    f.write_all(data).await?;
                    Ok(size as u32)
                }
                OpenLocation::Remote((node, peer)) => {
                    debug!("read from remote, node {}, fh {}, peer {}", node, fh, peer);
                    let now = SystemTime::now();
                    match self.remote.read().await.get(&peer) {
                        None => Err(FileSystemError::RemoteNotFoundError(peer.clone())),
                        Some(v) => {
                            debug!("read from remote, time use {}", now.elapsed().unwrap().as_nanos());
                            v.write(fh, offset, data).await
                        }
                    }
                }
            }
        } else {
            Err(FileSystemError::UnknownFileError(fh))
        }
    }

    pub async fn readdir(&self, inode: u64, _fh: u64, _offset: i64) -> Result<Vec<(u64, OsString, NodeAttr)>, FileSystemError> {
        let locate = self.metadata.read().await.children(inode).await;
        if let Some(vec) = locate {
            let mut array = Vec::new();
            for v in vec {
                let node = v.read().await;
                array.push((node.id(), node.name(), node.node_attr()))
            }
            Ok(array)
        } else {
            Err(FileSystemError::NodeNotFoundError(inode))
        }
    }

    pub async fn release(&self, fh: u64) -> Result<(), FileSystemError> {
        let location = self.opened.write().await.remove(&fh);
        match location {
            None => Ok(()),
            Some(location) => {
                match location {
                    OpenLocation::Local((node, file)) => {
                        let metadata = file.read().await.metadata().await?;
                        let attribute = Attribute::from(metadata);
                        let r = self.metadata.write().await.node_attribute(node, attribute).await?;
                        if let Some((path, attr)) = r {
                            let event = Event::Attribute{path, attr};
                            self.broadcast(event).await?;
                        }
                        Ok(())
                    },
                    OpenLocation::Remote((node, peer)) => {
                        trace!("release from remote {}, {}", node, peer);
                        if let Some(remote) = self.remote.read().await.get(&peer) {
                            remote.release(fh).await
                        } else {
                            warn!("remote not found for release {}, {}", fh, peer);
                            Ok(())
                        }
                    }
                }
            }
        }
    }

    pub async fn update(&self, path: &Path, attribute: Attribute) -> Result<(), FileSystemError> {
        self.metadata.write().await.path_attribute(path, attribute).await?;
        Ok(())
    }

    pub async fn filesystem(&self) -> Result<ReplyStatFs, FileSystemError> {
        let reply = self.local.storage_stat();
        Ok(reply)
    }
}